package person.tavion.springbootkafkademo;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.UUID;

@Component
public class KafkaConsumer1 {
    @Autowired
    private KafkaProducer1 kafkaProducer;

    @KafkaListener(topics = "myTopic", groupId = "myGroup")
    public void listen(String message) {
        System.out.println("消费中："+message);
        if (message.contains("test-1")) {
            // 异常触发,通过 KafkaConfig#DefaultErrorHandler 捕捉异常，并发送到死信队列（DeadLetterPublishingRecoverer）
            throw new RuntimeException("消费失败");
        }else if(message.contains("Transactional")){
            // 处理事务
            try {
                System.out.println("处理事务中。。。");
                Thread.sleep(5*60*1000);
                System.out.println("处理事务完成。。。");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        System.out.println("Received message in group myGroup: " + message);

    }

    @KafkaListener(id="myTopic.DLT", topics =  "myTopic.DLT", groupId = "myGroup.DLT")
    public void messListenerDLT(String message) {
        System.out.println("死信队列消费端 收到消息：" + message);
    }
}